package com.xiaoniu.mc.client.builder;

import com.xiaoniu.mc.client.annotation.XmqConsumer;
import com.xiaoniu.mc.client.config.BaseProperties;
import com.xiaoniu.mc.client.context.ApplicationContext;
import com.xiaoniu.mc.client.enums.ConsumeStrategy;
import com.xiaoniu.mc.client.exception.XmqConfigException;
import com.xiaoniu.mc.client.handler.XmqMessageHook;
import com.xiaoniu.mc.util.ClassScanUtil;
import com.xiaoniu.mc.util.MD5Util;
import com.xiaoniu.mc.util.PropertiesUtil;
import com.xiaoniu.mc.util.SerialNumberUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

import java.lang.annotation.Annotation;
import java.security.NoSuchAlgorithmException;
import java.util.List;

/**
 * @类描述：
 * @创建人：林继丰
 * @创建时间：2017/8/17 18:43
 * @版权：Copyright (c) 深圳市牛鼎丰科技有限公司-版权所有.
 */
public class MoudleScanner {

    /**
     * 扫描组件
     *
     * @throws IllegalAccessException
     * @throws InstantiationException
     * @throws MQClientException
     */
    public static void scanMoudle() throws IllegalAccessException, InstantiationException, MQClientException {
        String moudlePath = ApplicationContext.getBaseProperties().getMoudlePackage();
        if (StringUtils.isBlank(moudlePath))
            System.out.println("找不到模块扫描名称");
        List<Class<?>> clazzs = ClassScanUtil.scan(moudlePath);
        for (Class clazz : clazzs) {
            Annotation[] annotations = clazz.getAnnotations();
            for (Annotation annotation : annotations) {
                //如果是消费者注解
                if (annotation instanceof XmqConsumer) {
                    XmqMessageHook hook = (XmqMessageHook) clazz.newInstance();
                    initConsumer((XmqConsumer) annotation, hook);
                }
            }
        }
    }

    public static void scanConfig(String filePath) {
        PropertiesUtil.loadProperties(filePath);
        String nameServer = (String) PropertiesUtil.getProperties().get("xmq.nameserver");
        String moudlePackage = (String) PropertiesUtil.getProperties().get("xmq.moudle.package");
        if (StringUtils.isBlank(nameServer)) {
            throw new XmqConfigException("nameServer不存在", null);
        }
        BaseProperties properties = new BaseProperties();
        properties.setNameServer(nameServer);
        properties.setMoudlePackage(moudlePackage);
        ApplicationContext.setBaseProperties(properties);
    }

    /**
     * 生成消费者
     *
     * @param meta
     * @param hook
     * @throws MQClientException
     */
    private static void initConsumer(XmqConsumer meta, XmqMessageHook hook) throws MQClientException {
        try {
            validateConsumerGroup(meta);
        } catch (NoSuchAlgorithmException e) {
            e.printStackTrace();
            throw new XmqConfigException("初始化消费者失败", new Throwable());
        }
        String groupNames = meta.groups();
        String topic = meta.topic();
        String tags = meta.tags();
        String serialNum = SerialNumberUtil.getDigitRandom(10);
        String instanceName = String.format("%s_%s_%s", topic, tags, serialNum);

        boolean isOrderly = meta.isOrderLy();
        boolean isBroadcast = meta.isBroadcast();
        int maxThreadSum = meta.maxConsumerThreadSum();
        int minThreadSum = meta.minConsumerThreadSum();
        int batchCosumeMessageSum = meta.batchConsumerMessage();

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupNames);
        consumer.subscribe(topic, tags);
        if (isOrderly) {
            ConsumeStrategy.PUSH_MODEL_ORDERLY.registerHooks(consumer, hook);
        } else {
            ConsumeStrategy.PUSH_MODEL_CONCURRENCY.registerHooks(consumer, hook);
        }
        if (isBroadcast) {
            consumer.setMessageModel(MessageModel.BROADCASTING);
        }
        consumer.setNamesrvAddr(ApplicationContext.getBaseProperties().getNameServer());
        consumer.setInstanceName(instanceName);
        consumer.setConsumeThreadMax(maxThreadSum);
        consumer.setConsumeThreadMin(minThreadSum);
        consumer.setConsumeMessageBatchMaxSize(batchCosumeMessageSum);
        /**
         * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br>
         * 如果非第一次启动，那么按照上次消费的位置继续消费
         */
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.start();
        ApplicationContext.getTopicMapConsumer().put(instanceName, consumer);
        ApplicationContext.getTopicMaphooks().put(instanceName, hook);
        System.out.println("~启动服务消费者~" + instanceName);
    }


    private static void validateConsumerGroup(XmqConsumer meta) throws NoSuchAlgorithmException {
        String groupNames = meta.groups();
        String metaHashInfo = ApplicationContext.getGroupNameMapMeta().get(groupNames);
        String newMetaInfo = getConsumerMeta(meta);
        String hashNewMetaInfo = MD5Util.md5(newMetaInfo);
        if (StringUtils.isNotBlank(metaHashInfo) && !metaHashInfo.equals(hashNewMetaInfo)) {
            throw new XmqConfigException("存在相同的groupName" + groupNames + ",但是配置信息不一致!!", new Throwable());
        }
        ApplicationContext.getGroupNameMapMeta().put(groupNames, hashNewMetaInfo);
    }

    private static String getConsumerMeta(XmqConsumer meta) {
        StringBuilder sb = new StringBuilder();
        sb.append(meta.topic()).append(meta.tags()).append(meta.groups()).append(meta.isOrderLy())
                .append(meta.isBroadcast()).append(meta.minConsumerThreadSum())
                .append(meta.maxConsumerThreadSum()).append(meta.isEncrypt());
        return sb.toString();
    }
}
